package com.swak.reactivex.transport.udp;

import java.io.File;
import java.io.FileNotFoundException;
import java.net.InetAddress;
import java.net.InetSocketAddress;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.Constants;
import com.swak.exception.NoSupportException;
import com.swak.exception.RemotingException;
import com.swak.exception.ServerException;
import com.swak.reactivex.transport.ChannelHandler;
import com.swak.reactivex.transport.Codec;
import com.swak.reactivex.transport.Endpoint;
import com.swak.reactivex.transport.Message;
import com.swak.reactivex.transport.NettyChannel;
import com.swak.reactivex.transport.NettyClientHandler;
import com.swak.reactivex.transport.NettyCodec;
import com.swak.reactivex.transport.Peer;
import com.swak.reactivex.transport.codec.MultipleCodec;
import com.swak.reactivex.transport.resources.LoopResources;
import com.swak.registry.URL;
import com.swak.utils.Nets;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.DatagramChannel;

/**
 * UdpNettyPeer.
 * 
 * 没有调用 connect. 如果调用，则会与远端绑定，只能与指定远端交互数据.
 * 
 * 可以直接作为客户端和服务器端使用;
 * 有一个问题是，如果是多服务器发送数据，则只会使用单线程的方式；效率不高；
 * 
 * 如果只要作为客户端使用则可以使用 UdpNettyClient 来和具体的服务器端绑定；
 * 
 * @author DELL
 */
public class UdpNettyPeer extends Endpoint implements Peer {

	protected Logger logger = LoggerFactory.getLogger(UdpNettyPeer.class);

	private Channel channel;
	private Bootstrap bootstrap;
	private InetSocketAddress address;
	private final LoopResources resources;
	private boolean useMulticast = false;
	private boolean multicast = false;

	public UdpNettyPeer(URL url, LoopResources resources, ChannelHandler handler) throws RemotingException {
		super(url, handler);

		this.address = getUrl().toInetSocketAddress();
		this.resources = resources;

		if (url.getParameter(Constants.Use_Udp_Multicast, false)) {
			try {
				String Multicast_Host = url.getParameter(Constants.Multicast_Host_Key, Constants.Multicast_Host);
				int Multicast_Port = url.getParameter(Constants.Multicast_Port_Key, Constants.Multicast_Port);
				this.address = new InetSocketAddress(InetAddress.getByName(Multicast_Host), Multicast_Port);
				this.useMulticast = true;

			} catch (Throwable t) {
				throw new RemotingException("Failed to bind " + getClass().getSimpleName() + " on " + address
						+ ", cause: " + t.getMessage(), t);
			}
		}

		try {
			this.start();
			if (logger.isInfoEnabled()) {
				logger.info("Start " + getClass().getSimpleName() + " export " + address);
			}
		} catch (Throwable t) {
			throw new RemotingException(
					"Failed to bind " + getClass().getSimpleName() + " on " + address + ", cause: " + t.getMessage(),
					t);
		}
	}

	private void start() throws ServerException {
		if (this.started) {
			return;
		}
		this.started = true;
		bootstrap = new Bootstrap();
		bootstrap.option(ChannelOption.SO_REUSEADDR, Boolean.TRUE) //
				.option(ChannelOption.SO_BROADCAST, Boolean.TRUE)//
				.option(ChannelOption.IP_MULTICAST_LOOP_DISABLED, Boolean.TRUE)//
				.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);//

		// 配置： ServerBootstrap
		this.customBootstrap(bootstrap);

		bootstrap.handler(new ChannelInitializer<DatagramChannel>() {
			@Override
			public void initChannel(DatagramChannel ch) throws Exception {
				ChannelPipeline pipeline = ch.pipeline();

				// 配置： SocketChannel
				UdpNettyPeer.this.customChannel(ch);
				pipeline.addLast("handler", new NettyClientHandler(getUrl(), UdpNettyPeer.this));
			}
		});

		// 监听此端口
		ChannelFuture channelFuture = bootstrap.bind(this.address.getPort());
		channelFuture.syncUninterruptibly();
		this.channel = channelFuture.channel();

		// 加入组播
		if (this.useMulticast) {
			((DatagramChannel) this.channel).joinGroup(this.address, Nets.getLocalNetwork());
			this.multicast = true;
		}
	}

	@Override
	public void close() {
		if (logger.isInfoEnabled()) {
			logger.info("Close " + getClass().getSimpleName() + " bind " + address);
		}
		try {
			super.close();
		} catch (Throwable e) {
			logger.warn(e.getMessage(), e);
		}
		try {
			doClose();
		} catch (Throwable e) {
			logger.warn(e.getMessage(), e);
		}
	}

	protected void doClose() throws Throwable {
		try {
			if (channel != null) {
				channel.close();
			}
		} catch (Throwable e) {
			logger.warn(e.getMessage(), e);
		}

		try {
			if (this.useMulticast) {
				((DatagramChannel) this.channel).leaveGroup(this.address, Nets.getLocalNetwork());
			}

		} catch (Exception e) {
			logger.warn(e.getMessage(), e);
		}
	}

	/**
	 * 配置事件组，配置各种参数
	 * 
	 * @param bootstrap
	 */
	protected void customBootstrap(Bootstrap bootstrap) {
		bootstrap.group(resources.onServer()).channel(resources.onDatagramChannel());
	}

	private com.swak.reactivex.transport.Channel getChannel() {
		Channel c = channel;
		if (c == null) {
			return null;
		}
		return NettyChannel.getOrAddChannel(getUrl(), c, this);
	}

	/**
	 * 不需要向下传递： 连接事件
	 */
	@Override
	public void connected(com.swak.reactivex.transport.Channel channel) throws RemotingException {
		// super.connected(channel);
	}

	/**
	 * 不需要向下传递： 断开连接事件
	 */
	@Override
	public void disconnected(com.swak.reactivex.transport.Channel channel) throws RemotingException {
		// super.disconnected(channel);
	}

	/**
	 * 进行消息的转换
	 */
	@Override
	public void send(Object message, boolean sent) throws RemotingException {
		com.swak.reactivex.transport.Channel channel = getChannel();
		if (channel == null || !channel.isConnected()) {
			throw new RemotingException("message can not send, because channel is closed . url:" + getUrl());
		}

		// 封装请求
		Message request = (message instanceof Message) ? (Message) message : Message.forMessage(message);

		/*
		 * 如果是组播，则直接设置一个固定的地址
		 */
		if (this.useMulticast && request.getAddress() == null) {
			request.setAddress(this.address);
		}

		// 一定要检查：发送的地址不能为空
		if (request.getAddress() == null) {
			throw new IllegalArgumentException("must set address!");
		}

		/*
		 * 发送数据
		 */
		channel.send(message, sent);
	}

	@Override
	public void sendFile(File file, boolean sent) throws RemotingException, FileNotFoundException {
		throw new NoSupportException("un Support send file, you can use send Message");
	}

	@Override
	public boolean isConnected() {
		return !this.isClosed();
	}

	@Override
	public InetSocketAddress localAddress() {
		return address;
	}

	@Override
	public InetSocketAddress remoteAddress() {
		return address;
	}

	@Override
	public Object getAttribute(String key) {
		com.swak.reactivex.transport.Channel channel = getChannel();
		if (channel == null) {
			return null;
		}
		return channel.getAttribute(key);
	}

	@Override
	public void setAttribute(String key, Object value) {
		com.swak.reactivex.transport.Channel channel = getChannel();
		if (channel == null) {
			return;
		}
		channel.setAttribute(key, value);
	}

	@Override
	public void removeAttribute(String key) {
		com.swak.reactivex.transport.Channel channel = getChannel();
		if (channel == null) {
			return;
		}
		channel.removeAttribute(key);
	}

	@Override
	public boolean hasAttribute(String key) {
		com.swak.reactivex.transport.Channel channel = getChannel();
		if (channel == null) {
			return false;
		}
		return channel.hasAttribute(key);
	}

	@Override
	public boolean isUnicast() {
		return !this.multicast;
	}

	/**
	 * 配置编解码协议
	 * 
	 * @param ch
	 */
	protected void customChannel(DatagramChannel ch) {
		ChannelPipeline pipeline = ch.pipeline();
		Codec codec = new MultipleCodec(getUrl());
		NettyCodec nettyCodec = new NettyCodec(codec, getUrl(), this);
		pipeline.addLast(NettyCodec.Encoder_Name, nettyCodec.getEncoder());
		pipeline.addLast(NettyCodec.Decoder_Name, nettyCodec.getDecoder());
	}
}
